package com.lesserpand.rabbitmq.rabbitmq.consumer;

import com.lesserpand.rabbitmq.rabbitmq.bean.OrderDO;
import com.lesserpand.rabbitmq.rabbitmq.bean.dto.OrderDTO;
import com.lesserpand.rabbitmq.rabbitmq.bean.dto.OrderItemDTO;
import com.lesserpand.rabbitmq.rabbitmq.constant.MqEnum;
import com.lesserpand.rabbitmq.rabbitmq.provider.ProviderUtils;
import com.lesserpand.rabbitmq.rabbitmq.service.OrderService;
import com.rabbitmq.client.Channel;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

/**
 * @author pand
 */
@Component
public class ConsumerUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerUtils.class);

    @Autowired
    private OrderService orderService;

    @Autowired
    private ProviderUtils providerUtils;

    /**
     * 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息。@RabbitListener不实现SimpleMessageListenerContainer的情况下，使用默认的消息处理，当实现了之后，需要自己手动处理消息。如：
     * RabbitMqConfig当中的@Bean("messageListenerContainer")自己实现了消息处理。
     * channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     * <p>
     * ack返回false，并重新回到队列，api里面解释得很清楚
     * channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
     * <p>
     * 拒绝消息
     * channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
     *
     * @param message     :解码后的消息
     * @param deliveryTag :使用@Header接口获取messageProperties中的DELIVERY_TAG属性。
     * @param channel     : 接受消息所使用的信道
     */
    @RabbitListener(queues = {"first-queue", "second-queue"}, containerFactory = "myRabbitListenerContainerFactory")
    // @RabbitListener(queues = {"first-queue", "second-queue"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
        // 手动应答
        channel.basicAck(deliveryTag, false);
        System.out.println("FirstConsumer {} handleMessage :" + message);
    }

    /**
     * 延迟队列实际消费队列监听
     *
     * @param message
     * @throws Exception
     */
    @RabbitListener(queues = {"message.center.create"}, containerFactory = "rabbitListenerContainerFactory")
    public void ttlHandleMessage(String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
        // 控制处理消息，达到流量控制
        channel.basicQos(100);
        // 订单id
        Long orderId = Long.valueOf(message);
        // 根据订单id查询订单信息
        OrderDO orderDO = orderService.findById(orderId);
        if (null == orderDO) {
            throw new RuntimeException("订单信息不存在！");
        }
        Byte status = orderDO.getOrderStatus();
        switch (status) {
            case 1:
                // 1.5个小时以后，订单仍未支付，发送短信提醒用户3个小时未支付，订单将自动取消，然后将订单重新放入延迟队列
                LOGGER.info("亲爱的用户您好，您有订单未支付，如果订单超过3个小时未支付，订单将自动取消，请及时支付订单！");
                channel.basicAck(deliveryTag, false);
                // 重新放入ttl
                providerUtils.sendMessage(orderId, MqEnum.TTL_DIRECT_EXCHANGE.getValue(), MqEnum.MESSAGE_TTL_QUEUE2_KEY.getValue(), 30000);
                break;
            case 2:
                // 订单已经支付，直接消费掉
                LOGGER.info("订单已经支付！");
                channel.basicAck(deliveryTag, false);
                break;
            case 3:
                break;
            case 4:
                break;
            default:
                throw new RuntimeException("订单状态存在问题！");
        }
        // 处理消息
        System.out.println("FirstConsumer {} ttlHandleMessage :" + message + "消费实际：" + LocalDateTime.now());
    }

    /**
     * 延迟队列实际消费队列监听
     *
     * @param message
     * @throws Exception
     */
    @RabbitListener(queues = {"message.center.create2"}, containerFactory = "rabbitListenerContainerFactory")
    public void ttlHandleMessage2(String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
        // 控制处理消息，达到流量控制
        channel.basicQos(100);
        // 订单id
        Long orderId = Long.valueOf(message);
        // 根据订单id查询订单信息
        OrderDO orderDO = orderService.findById(orderId);
        if (null == orderDO) {
            throw new RuntimeException("订单信息不存在！");
        }
        Byte status = orderDO.getOrderStatus();
        switch (status) {
            case 1:
                // 1.5个小时以后，订单仍未支付，发送短信提醒用户3个小时未支付，订单将自动取消，然后将订单重新放入延迟队列
                LOGGER.info("亲爱的用户您好，您的订单超过3个小时未支付，订单自动取消！");
                OrderDTO dto = new OrderDTO();
                dto.setId(orderId);
                dto.setOrderStatus(new Byte("0"));
                // 取消订单
                orderService.update(dto);
                channel.basicAck(deliveryTag, false);
                break;
            case 2:
                // 订单已经支付，直接消费掉
                LOGGER.info("订单已经支付！");
                channel.basicAck(deliveryTag, false);
                break;
            case 3:
                break;
            case 4:
                break;
            default:
                throw new RuntimeException("订单状态存在问题！");
        }
        // 处理消息
        System.out.println("FirstConsumer {} ttlHandleMessage :" + message + "消费实际：" + LocalDateTime.now());
    }

    /**
     * 监听抢购订单队列，进行订单入库以及放入延迟队列，判断是否支付
     *
     * @param message
     * @throws Exception
     */
    @RabbitListener(queues = {"skill.message.queue"}, containerFactory = "rabbitListenerContainerFactory")
    public void skillHandleMessage(String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
        // 控制处理消息，达到流量控制
        channel.basicQos(10);
        // 用户id与商品id
        String userAndProId = String.valueOf(message);
        if (StringUtils.isEmpty(userAndProId)) {
            throw new RuntimeException("抢购信息有问题！");
        }
        String[] ids = StringUtils.split(userAndProId, ",");
        if (ids.length != 2) {
            throw new RuntimeException("抢购信息有问题！");
        }
        Long userId = Long.valueOf(ids[0]);
        Long proId = Long.valueOf(ids[1]);
        OrderDTO dto = new OrderDTO();
        dto.setUserId(userId);
        dto.setOrderStatus(new Byte("1"));
        dto.setTotalPrice(BigDecimal.valueOf(111));

        List<OrderItemDTO> list = new ArrayList<>();
        OrderItemDTO orderItemDTO = new OrderItemDTO();
        orderItemDTO.setProId(proId);
        orderItemDTO.setItemName("商品名称");
        orderItemDTO.setItemNum(1);
        orderItemDTO.setPrice(BigDecimal.valueOf(111));
        list.add(orderItemDTO);
        dto.setList(list);
        orderService.save(dto);
        channel.basicAck(deliveryTag, false);
        // 处理消息
        System.out.println("FirstConsumer {} ttlHandleMessage :" + message + "消费实际：" + LocalDateTime.now());
    }

}
